package com.lesent.service.business;

import com.lesent.hadoop.domain.OrderBean;
import com.lesent.service.TestOrderMapReducer;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat;

import java.io.IOException;

/**
 * @author shengtao
 * @Description: track_log的省份数据清洗
 * @date 2019/01/07 12:57
 */
public class TrackLogMapReducer {

    class LogMapper extends Mapper<LongWritable,Text,LongWritable,Text> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] conts = line.split("\t");

            //长度不够，过滤
            if(conts.length < 32){
                context.getCounter("Web Data ETL","Data lenght is too shart").increment(1);
                return;
            }

            String url = conts[0];
            if(StringUtils.isBlank(url)){
                context.getCounter("Web Data ETL","Url is blank").increment(1);
                return;
            }

            String province = conts[20];
            Integer provinceId = Integer.MAX_VALUE;

            try {
                provinceId = Integer.valueOf(province);
            }catch (Exception e){
                context.getCounter("Web Data ETL","ProvinceId is Unknown").increment(1);
                return;
            }

            if(StringUtils.isBlank(province) || provinceId == Integer.MAX_VALUE){
                context.getCounter("Web Data ETL","ProvinceId is Unknown");
                return;
            }

            context.write(key,value);
        }
    }

    class LogReducer extends Reducer<LongWritable ,Text,Text,NullWritable>{
        @Override
        protected void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            for (Text value:values){
                context.write(value,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) throws Exception {
        //创建hadoop配置
        String nameNodeUrl = "hdfs://192.168.220.135:9000";
        String nameNodeName = "fs.defaultFS";
        Configuration configuration = new Configuration();
        configuration.set(nameNodeName, nameNodeUrl);

        String inputPath = "/user/hive/warehouse/db_web_data.db/track_log/date1=20150828/hour1=18/2015082818"; //计算文件路径
        String outPath = "/web_etl/2015082818"; //输出文件路径

        Job job = Job.getInstance(configuration);
        job.setJarByClass(TrackLogMapReducer.class); //设置程序jar包所在的路径

        FileInputFormat.addInputPath(job,new Path(inputPath));
        //设置mapper和reducer处理类
        job.setMapperClass(LogMapper.class);
        //设置mapper和reducer的key和value的类型
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(LogReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileOutputFormat.setOutputPath(job,new Path(outPath));
        //同事指定分区数量的reduceTask
//        job.setNumReduceTasks(1);
        //执行job同步
        boolean result = job.waitForCompletion(true);

        if (!result) {
            System.out.println("track log task fail!!!");
        }
    }

}
